from kafka import KafkaProducer
import json
import time
import random
import logging

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def generate_access_log():
    """生成模拟的JSON访问日志"""
    log_data = {
        "ip": f"{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}",
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "method": random.choice(["GET", "POST", "PUT", "DELETE"]),
        "path": f"/{random.choice(['api','data','user','images'])}/{random.randint(1000,9999)}",
        "status": random.choice([200, 201, 400, 401, 403, 404, 500]),
        "response_time": round(random.uniform(0.05, 2.0), 3),
        "user_agent": random.choice([
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
            "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/",
            "Safari/537.36",
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
        ])
    }
    return log_data  # 取消注释这行

def main():
    # Kafka 配置
    KAFKA_BROKERS = "kafka-log.data-service.svc.cluster.local:9092"  # Kafka 集群地址，多个用逗号分隔

    KAFKA_TOPIC = "access-logs"        # Kafka 主题名称
    
    # 创建生产者实例
    try:
        producer = KafkaProducer(
            bootstrap_servers=KAFKA_BROKERS,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            acks='all',  # 确保消息被完整写入
            retries=3,   # 发送失败时的重试次数
            api_version=(0, 10, 1)  # 指定Kafka API版本，避免自动检测失败
        )
    except Exception as e:
        logging.error(f"Failed to create Kafka producer: {e}")
        logging.error("This might be due to an outdated kafka-python library. Try upgrading it with: pip install --upgrade kafka-python")
        return

    logging.info(f"Started Kafka producer connected to {KAFKA_BROKERS}")

    try:
        # 持续生成并发送日志
        while True:
            log_entry = generate_access_log()
            
            # 发送消息
            future = producer.send(KAFKA_TOPIC, value=log_entry)
            
            # 添加回调来处理发送结果
            def on_send_success(record_metadata):
                logging.debug(f"Message sent successfully to topic {record_metadata.topic}, partition {record_metadata.partition}, offset {record_metadata.offset}")
            
            def on_send_error(excp):
                logging.error(f"Failed to send message: {excp}")
            
            future.add_callback(on_send_success)
            future.add_errback(on_send_error)
            
            logging.info(f"Sent log: {json.dumps(log_entry)}")
            
            # 模拟实时日志（每秒1-5条）
            time.sleep(random.uniform(0.2, 1.0))
            
    except KeyboardInterrupt:
        logging.info("Stopping producer...")
    except Exception as e:
        logging.error(f"Error occurred: {e}")
    finally:
        # 确保所有消息完成发送
        producer.flush()
        producer.close()
        logging.info("Producer closed")

if __name__ == "__main__":
    main()